package rabbitmq.publisher.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import rabbitmq.Connect;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublisherConfirm {
    public static final int MESSAGE_COUNT = 20000; // 发送消息的个数

    public static void main(String[] args) throws Exception {
        // Publishing Messages Individually 单独确认
//        publishingMessagesIndividually();
        // Publishing Messages in Batches 批量确认
//        publishingMessagesInBatches();
        // Handling Publisher Confirms Asynchronously 异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    // 单独确认 耗时长
    private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = Connect.createConnection()) {
            // 创立信道
            Channel channel = connection.createChannel();
            // 开启确认
            channel.confirmSelect();
            // 声明
            channel.queueDeclare(Constants.PUBLISHER_CONFIRM_QUEUE1, true, false, false, null);

            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirm..." + i;
                // 发布
                channel.basicPublish("", Constants.PUBLISHER_CONFIRM_QUEUE1, null, msg.getBytes());
                // 等待确认
                channel.waitForConfirmsOrDie(3000);
            }
            long end = System.currentTimeMillis();
            System.out.println("单独确认发送 "+ MESSAGE_COUNT +" 个消息的耗时为 " + (end - start) + "ms");
            channel.close();
        }
    }

    // 批量确认 耗时短 但是处理大量数据时耗时比异步确认长
    private static void publishingMessagesInBatches() throws Exception {
        try (Connection connection = Connect.createConnection()) {
            // 创立信道
            Channel channel = connection.createChannel();
            // 开启确认
            channel.confirmSelect();
            // 声明
            channel.queueDeclare(Constants.PUBLISHER_CONFIRM_QUEUE2, true, false, false, null);

            long start = System.currentTimeMillis();
            // 设置阈值
            int maxSize = 100;
            int curSize = 0;

            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirm..." + i;
                channel.basicPublish("", Constants.PUBLISHER_CONFIRM_QUEUE2, null, msg.getBytes());
                curSize++;
                if (curSize == maxSize) {
                    // 确认
                    channel.waitForConfirmsOrDie(3000);
                    curSize = 0;
                }
            }
            if (curSize > 0) {
                // 如果有 那么就还要确认
                channel.waitForConfirmsOrDie(3000);
            }
            long end = System.currentTimeMillis();
            System.out.println("批量确认发送 "+ MESSAGE_COUNT +" 个消息的耗时为 " + (end - start) + "ms");

        }
    }

    // 异步确认 耗时最短
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {
        try (Connection connection = Connect.createConnection()) {
            // 创立信道
            Channel channel = connection.createChannel();
            // 开启确认
            channel.confirmSelect();
            // 声明
            channel.queueDeclare(Constants.PUBLISHER_CONFIRM_QUEUE3, true, false, false, null);

            long start = System.currentTimeMillis();

            // 存储没有确认的DeliveryTag集合
            SortedSet<Long> confirmNo = Collections.synchronizedSortedSet(new TreeSet<>());

            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    // 根据 multiple 清除
                    if (multiple) {
                        confirmNo.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    // 根据 multiple 清除
                    if (multiple) {
                        confirmNo.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmNo.remove(deliveryTag);
                    }

                    // 再根据业务来觉得错误的情况下应该怎么处理
                }
            });
            // 发送
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirm..." + i;
                confirmNo.add(channel.getNextPublishSeqNo()); // 将 Tag添加到集合中
                channel.basicPublish("", Constants.PUBLISHER_CONFIRM_QUEUE3, null, msg.getBytes());
            }
            // 阻塞等待 直到它都处理好所有发送消息
            while (!confirmNo.isEmpty()) Thread.sleep(10);
            long end = System.currentTimeMillis();
            System.out.println("批量确认发送 "+ MESSAGE_COUNT +" 个消息的耗时为 " + (end - start) + "ms");

        }
    }
}
